package consumer

import (
	"context"
	"encoding/json"
	"fmt"
	"knowledge_admin/internal/application/usecase"
	"knowledge_admin/internal/domain"
	"knowledge_admin/pkg/adapters/queue/pubsub"
	"knowledge_admin/pkg/adapters/queue/pubsub/pulsar"
	"knowledge_admin/pkg/framework"
	v1 "knowledge_admin/proto/api/v1"
	"strconv"
)

type PulsarConsumerService struct {
	logger       framework.Logger
	pulsar       *pulsar.PulsarClient
	goodsUseCase *usecase.GoodsUseCase
}

func NewPulsarConsumerService(logger framework.Logger, pulsar *pulsar.PulsarClient, goodsUseCase *usecase.GoodsUseCase) *PulsarConsumerService {
	return &PulsarConsumerService{
		logger:       logger,
		pulsar:       pulsar,
		goodsUseCase: goodsUseCase,
	}
}

// Start 实现 framework.Server 接口
func (s *PulsarConsumerService) Start() error {
	go s.startConsumer()
	return nil
}

// Stop 实现 framework.Server 接口
func (s *PulsarConsumerService) Stop() error {
	// 实现优雅关闭逻辑
	return nil
}

// Serve 实现 framework.Server 接口
func (s *PulsarConsumerService) Serve(ctx context.Context) error {
	// 由于我们在 Start() 中已经启动了消费者协程
	// Serve 方法可以简单返回 nil
	err := s.Start()
	if err != nil {
		return err
	}
	return nil
}

func (s *PulsarConsumerService) startConsumer() {
	_ = s.pulsar.GetSubscriber("shop_goods_sync").Start(context.Background(), func(ctx context.Context, msg pubsub.Message) error {
		s.logger.Info("msg", msg.Properties())
		s.logger.Info("Received message", msg.Properties())

		// 解析消息内容
		var payload struct {
			Shops []*v1.SyncGoodsShop `json:"shops"`
		}
		err := json.Unmarshal(msg.Payload(), &payload)
		if err != nil {
			s.logger.Error("Failed to unmarshal message payload", err)
			return err
		}

		// 转换 shops 为 domain.Shop 类型
		domainShops := make([]*domain.Shop, 0, len(payload.Shops))
		for _, syncShop := range payload.Shops {
			shop := syncShop.GetShop()
			if shop != nil {
				domainShops = append(domainShops, &domain.Shop{
					Id:         shop.GetId(),
					TenantId:   msg.Properties()["tenant_id"],
					Platform:   shop.GetPlatform(),
					ShopName:   shop.GetShopName(),
					PlatUserId: shop.GetPlatUserId(),
				})
			}
		}

		onShelves, _ := strconv.ParseBool(msg.Properties()["on_shelves"])
		containedAttr := payload.Shops[0].ContainedAttr
		err = s.goodsUseCase.SyncGoodsByShopIDs(
			ctx,
			domainShops,
			msg.Properties()["customer_id"],
			msg.Properties()["knowledge_base_id"],
			onShelves,
			containedAttr,
		)
		if err != nil {
			s.logger.Error("Sync Error", err)
			return err
		}
		// 确认消息已处理
		err = msg.Ack()
		if err != nil {
			s.logger.Error("ack error", err)
		}
		return nil
	})
	fmt.Printf("你好")
}
